package migrator

import (
	"context"
	"fmt"
	"strings"

	"gitlab.bangdao-tech.com/ops/ck-migrator/pkg/config"

	"gitlab.bangdao-tech.com/ops/ck-migrator/pkg/clickhouse"

	"github.com/riete/errors"
)

type MigrateSqlBuilder struct {
	SourceDatabase   string
	SourceSortingKey string
	TargetDatabase   string
	Table            config.Table
}

func (m MigrateSqlBuilder) Build(batchCount int, partitionKey, partitionValue string) []string {
	var sqls []string
	for i := 0; i < batchCount; i++ {
		var s string
		// 当 batchCount 为 1 且 batchSize 为 0 时，不使用 farmHash64 进行数据分片
		if batchCount == 1 && config.Config.BatchSize == 0 {
			s = fmt.Sprintf(
				"INSERT INTO %s.%s%s SELECT %s FROM remote('%s:%d', '%s.%s', '%s', '%s')",
				m.TargetDatabase,
				m.Table.Target,
				m.Table.TargetColumn(),
				m.Table.SourceColumn(),
				config.Config.Source.Host,
				config.Config.Source.Port,
				m.SourceDatabase,
				m.Table.Source,
				config.Config.Source.Username,
				config.Config.Source.Password,
			)
		} else {
			s = fmt.Sprintf(
				"INSERT INTO %s.%s%s SELECT %s FROM remote('%s:%d', '%s.%s', '%s', '%s') WHERE farmHash64(%s) %% %d = %d",
				m.TargetDatabase,
				m.Table.Target,
				m.Table.TargetColumn(),
				m.Table.SourceColumn(),
				config.Config.Source.Host,
				config.Config.Source.Port,
				m.SourceDatabase,
				m.Table.Source,
				config.Config.Source.Username,
				config.Config.Source.Password,
				m.SourceSortingKey,
				batchCount,
				i,
			)
		}
		
		// 处理条件和分区条件
		hasWhereClause := (batchCount > 1 || config.Config.BatchSize > 0) // 当使用分片时，SQL已经有WHERE子句
		if m.Table.Condition != "" {
			// 如果 batchSize 为 0 且 batchCount 为 1，则需要添加 WHERE 子句
			if batchCount == 1 && config.Config.BatchSize == 0 {
				s = fmt.Sprintf("%s WHERE %s", s, m.Table.Condition)
			} else {
				s = fmt.Sprintf("%s AND %s", s, m.Table.Condition)
			}
			hasWhereClause = true
		}
		if partitionKey != "" && partitionValue != "" {
			if !hasWhereClause {
				if strings.HasPrefix(partitionValue, "(") {
					s = fmt.Sprintf("%s WHERE %s=%s", s, partitionKey, partitionValue)
				} else {
					s = fmt.Sprintf("%s WHERE %s='%s'", s, partitionKey, partitionValue)
				}
			} else {
				if strings.HasPrefix(partitionValue, "(") {
					s = fmt.Sprintf("%s AND %s=%s", s, partitionKey, partitionValue)
				} else {
					s = fmt.Sprintf("%s AND %s='%s'", s, partitionKey, partitionValue)
				}
			}
		}
		sqls = append(sqls, s)
	}
	return sqls
}

type Migrator struct {
	source *clickhouse.Clickhouse
}

func (m Migrator) DetermineBatchCount(dbname, tableName, partitionKey, partitionValue, condition string) (uint64, uint64, errors.Error) {
	// 确保数据库名和表名不包含特殊字符，避免SQL注入和语法错误
	dbname = strings.Trim(dbname, " '\"()[]")
	tableName = strings.Trim(tableName, " '\"()[]")

	// 清理分区键，但保留分区值中的括号（如果有）
	if partitionKey != "" {
		partitionKey = strings.Trim(partitionKey, " '\"[]")
	}

	sql := fmt.Sprintf(`SELECT count() AS count FROM %s.%s`, dbname, tableName)
	if partitionKey != "" && partitionValue != "" {
		if strings.HasPrefix(partitionValue, "(") {
			sql = fmt.Sprintf("%s WHERE %s=%s", sql, partitionKey, partitionValue)
		} else {
			sql = fmt.Sprintf("%s WHERE %s='%s'", sql, partitionKey, partitionValue)
		}
	}
	if condition != "" {
		if strings.TrimSpace(condition) != "" {
			if !strings.Contains(sql, "WHERE") {
				sql = fmt.Sprintf("%s WHERE %s", sql, condition)
			} else {
				sql = fmt.Sprintf("%s AND %s", sql, condition)
			}
		}
	}
	row := m.source.QueryRow(context.Background(), sql)
	var count uint64
	if err := row.Scan(&count); err != nil {
		return count, 0, errors.NewFromErr(err).Trace("查询count失败").Trace(sql)
	}
	
	// 当 batchSize 为 0 时，表示不限制行数，批次固定为 1 批
	if config.Config.BatchSize == 0 {
		return count, 1, nil
	}
	
	if count < config.Config.BatchSize {
		return count, 1, nil
	}
	return count, count / config.Config.BatchSize, nil
}

func (m Migrator) GetPartitionAndSortingKey(dbname, tableName string) (string, string, errors.Error) {
	// 确保数据库名和表名不包含特殊字符，避免SQL注入和语法错误
	dbname = strings.Trim(dbname, " '\"()[]")
	tableName = strings.Trim(tableName, " '\"()[]")
	sql := fmt.Sprintf("select partition_key AS partitionKey, sorting_key AS sortingKey from system.`tables` t where `database` = '%s' and `name` = '%s' and is_temporary = 0", dbname, tableName)
	row := m.source.QueryRow(context.Background(), sql)
	var partitionKey string
	var sortingKey string
	err := row.Scan(&partitionKey, &sortingKey)
	if err != nil {
		return partitionKey, sortingKey, errors.NewFromErr(err).Trace("查询分区key失败").Trace(sql)
	}
	return partitionKey, sortingKey, nil
}

func (m Migrator) GetDistributedLocalTableName(dbname, tableName string) (string, errors.Error) {
	// 确保表名不包含特殊字符，避免SQL注入和语法错误
	tableName = strings.Trim(tableName, " '\"()[]")
	sql := fmt.Sprintf("select engine_full AS engineFull from `system`.`tables` t where `database` ='%s' and `name` ='%s' and  is_temporary = 0 and `engine` = 'Distributed'", dbname, tableName)
	row := m.source.QueryRow(context.Background(), sql)
	var engineFull string
	if err := row.Scan(&engineFull); err != nil {
		return "", errors.NewFromErr(err).Trace("查询本地分区表失败").Trace(sql)
	}

	// 更安全地提取本地表名
	// ClickHouse Distributed引擎格式通常为: Distributed(cluster, database, table, sharding_key)
	// 我们需要提取第三个参数，即本地表名

	// 首先尝试使用更可靠的方法解析引擎字符串
	startIdx := strings.Index(engineFull, "(")
	if startIdx == -1 {
		return "", errors.New("无法从 engine_full 中提取本地表名，缺少左括号: " + engineFull)
	}

	// 提取括号内的内容并按逗号分割
	paramsStr := engineFull[startIdx+1:]
	endIdx := strings.LastIndex(paramsStr, ")")
	if endIdx == -1 {
		return "", errors.New("无法从 engine_full 中提取本地表名，缺少右括号: " + engineFull)
	}

	paramsStr = paramsStr[:endIdx]

	// 使用更可靠的方法分割参数，考虑到参数中可能包含逗号
	var params []string
	inQuote := false
	inBracket := 0
	start := 0

	for i, c := range paramsStr {
		if c == '\'' && (i == 0 || paramsStr[i-1] != '\\') {
			inQuote = !inQuote
		} else if !inQuote {
			if c == '(' {
				inBracket++
			} else if c == ')' {
				inBracket--
			} else if c == ',' && inBracket == 0 {
				params = append(params, strings.TrimSpace(paramsStr[start:i]))
				start = i + 1
			}
		}
	}

	// 添加最后一个参数
	params = append(params, strings.TrimSpace(paramsStr[start:]))

	if len(params) < 3 {
		return "", errors.New("无法从 engine_full 中提取本地表名，参数不足: " + engineFull)
	}

	// 获取本地表名（第三个参数）并清理引号和括号
	localTableName := params[2]
	localTableName = strings.Trim(localTableName, " '\"")

	return localTableName, nil
}

func (m Migrator) GetPartitionList(dbname, tableName string, reverseOrder bool) ([]string, bool, errors.Error) {
	// 确保数据库名和表名不包含特殊字符，避免SQL注入和语法错误
	dbname = strings.Trim(dbname, " '\"()[]")
	tableName = strings.Trim(tableName, " '\"()[]")

	// 确保集群名称也经过清理
	clusterName := strings.Trim(config.Config.Source.ClusterName, " '\"()[]")

	orderBy := "`partition`"
	if reverseOrder {
		orderBy = "`partition` DESC"
	}
	sql := fmt.Sprintf("select distinct `partition`, partition_id from cluster(%s, `system`.parts) p where `database` = '%s' and `table` = '%s' and active = 1 order by %s", clusterName, dbname, tableName, orderBy)
	var result []struct {
		Partition   string `ch:"partition"`
		PartitionId string `ch:"partition_id"`
	}
	if err := m.source.Select(context.Background(), &result, sql); err != nil {
		return nil, false, err.Trace("查询partition list失败").Trace(sql)
	}
	var partitions []string
	isPartition := true
	if len(result) == 1 && result[0].PartitionId == "all" {
		isPartition = false
	}
	for _, r := range result {
		partitions = append(partitions, r.Partition)
	}
	return partitions, isPartition, nil
}

func New(source *clickhouse.Clickhouse) *Migrator {
	return &Migrator{source: source}
}
